package com.dmp.total

/**
  * 需求 3.1：日志转Parquet文件
  * 运行参数：C:\Users\Administrator\Desktop\x\data.txt C:\Users\Administrator\Desktop\x\a snappy
  */

import com.dmp.beans.Logs
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
  * 1）要求一：将数据转换成parquet文件格式
  * 2）要求二：序列化方式采用KryoSerializer方式
  * 3）要求三：parquet文件采用snappy压缩方式
  *
  * 思路：先把文件变成rdd再变成DataFrame，然后通过df.write.format("parquet")
  * HDFS txt   -> parquet
  * var rdd=sc.textText("xxx")
  * rdd -> DataFrame
  * RowRDD+scahame  或
  * RDD[Log].toDF
  * df.write.format("parquet")
  */
object Txt2Parquet {
  def main(args: Array[String]): Unit = {
    /**
      * 第一步：判断参数是否符合需求
      * 原始的文件路径 输出的文件路径 压缩格式
      */
    if (args.length < 3) {
      println(
        """
          |com.dmp.total.Txt2Parquet <dataPath> <outputPath> <compressionCode>
          |<dataPath>:日志所在的路径
          |<outputPath>：结果文件存放的路径
          |<compressionCode>：指定的压缩格式
        """.stripMargin)
      System.exit(0)
    }
    /**
      * 第二步：接收参数
      */
    val Array(dataPath, outputPath, compressionCode) = args
    /**
      * 第三步：创建SparkSession对象
      */
    val conf = new SparkConf()
    //设置序列化的格式
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.setMaster("local")
    //注册Logs类的序列化格式为Kryo
    conf.registerKryoClasses(Array(classOf[Logs]))
    //指定压缩格式
    conf.set("spark.io.compression.codec", compressionCode)
    val spark = SparkSession.builder()
      .config(conf)
      .getOrCreate()

    import spark.implicits._

    /**
      * 第四步：读取文件，对文件做相对应的操作
      */
    val logRDD: RDD[Logs] = spark.sparkContext.textFile(dataPath).map(line => Logs.line2Log(line))

    //   val df = spark.createDataFrame(logRDD)

    val df = logRDD.toDF()

    /**
      * 第五步：指定文件存放的位置
      */
    df.write.parquet(outputPath)

    spark.stop()

  }

}
